package com.hmall.common.utils;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.UUID;

/**
 * @author gonghuaiguo
 */
@Slf4j
@RequiredArgsConstructor
public class RabbitMqHelper {

    private final RabbitTemplate rabbitTemplate;

    /**
     * 发送普通消息
     *
     * @param exchange
     * @param routingkey
     * @param msg
     */
    public void sendMessage(String exchange, String routingkey, Object msg) {
        log.debug("准备发送消息，exchange:{},routingKey:{},msg:{}", exchange, routingkey, msg);
        // 服务间传递用户信息

        rabbitTemplate.convertAndSend(exchange, routingkey, msg,message -> {
            message.getMessageProperties().setHeader("user-info",UserContext.getUser());
            log.info("准备发送消息，获取用户信息userInfo:{}",UserContext.getUser());
            return message;
        });
    }

    /**
     * 发送超时消息
     *
     * @param exchange
     * @param routingkey
     * @param msg
     * @param delay
     */
    public void sendDelayMessage(String exchange, String routingkey, Object msg, int delay) {
        rabbitTemplate.convertAndSend(exchange, routingkey, msg, message -> {
            message.getMessageProperties().setHeader("user-info",UserContext.getUser());
            log.info("远程发送 获取用户信息:{}",UserContext.getUser());
            message.getMessageProperties().setDelay(delay);
            return message;
        });
    }


    public void sendMessageWithConfirm(String exchange, String routingKey, Object msg, int maxRetries) {
        log.debug("准备发送消息，exchange:{},routingKey:{},msg:{}", exchange, routingKey, msg);
        CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
        cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {

            int retryCount;

            @Override
            public void onFailure(Throwable ex) {
                log.error("处理ack回执失败:", ex);
            }

            @Override
            public void onSuccess(CorrelationData.Confirm result) {
                if (result != null && !result.isAck()) {
                    log.debug("消息发送失败，收到nack，已重试次数:{}",retryCount);
                    if(retryCount >= maxRetries) {
                        log.error("消息发送重试次数耗尽，发送失败");
                        return;
                    }
                }
                CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
                cd.getFuture().addCallback(this);
                rabbitTemplate.convertAndSend(exchange,routingKey,msg,cd);
                retryCount++;
            }
        });
        rabbitTemplate.convertAndSend(exchange,routingKey,msg,cd);
    }
}
